package com.hrt.hudi.spark.hive

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hive.MultiPartKeysValueExtractor
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * spark通过hudi的Mor模式更新写入hdfs并映射到hive表
 */
object SparkHudiMorUpdateMapHive {
  def main(args: Array[String]): Unit = {

    val sparkSession = SparkSession.builder()
      .appName("sparkhudi")
      .master("local")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()
    import sparkSession.implicits._
    val updateDataDF = sparkSession.read.option("header","true")
      .csv("file:\\F:\\P8\\workspace\\bigdatalake\\data\\sparkhudiUpdate.csv").as[PersonInfo]

    //向Hudi 更新数据，注意，必须指定Hive对应配置
    updateDataDF.write.format("org.apache.hudi")
      //设置写出模式，默认就是COW
      .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
      .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
      .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
      .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"data_dt")
      .option("hoodie.insert.shuffle.parallelism", "2")
      .option("hoodie.upsert.shuffle.parallelism", "2")
      .option(HoodieWriteConfig.TABLE_NAME, "person_infos2")
      //关于Hive设置
      //指定HiveServer2 连接url
      .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://hadoop102:10000")
      //指定Hive 对应的库名
      .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
      //指定Hive映射的表名称
      .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"infos2")
      //Hive表映射对的分区字段
      .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"data_dt")
      //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
      .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
      //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
      .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
      .mode(SaveMode.Append)
      .save("/hudi/spark/person_infos2")

  }
}
